package git.soulbgm.job;

import java.util.*;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import git.soulbgm.common.enums.Status;
import git.soulbgm.common.okhttp.OkHttpUtil;
import git.soulbgm.config.SysConfig;
import git.soulbgm.pojo.*;
import git.soulbgm.pojo.dto.*;
import git.soulbgm.service.AlarmInfoReport;
import git.soulbgm.service.AlarmRuleInfoService;
import git.soulbgm.service.NodeInfoService;
import git.soulbgm.service.NodeUsedInfoService;
import git.soulbgm.utils.*;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.Response;
import okhttp3.ResponseBody;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

import static git.soulbgm.common.MonitorConstants.*;

/**
 * 设备节点运行情况收集任务
 *
 * @author SoulBGM
 * @date 2021/01/29
 */
@Service
public class NodeUsedPullTask {

    private static final LogHelper LOG = new LogHelper(NodeHealthMonitorTask.class);
    private static final String RULE_TYPE = RULE_TYPE_NODE;
    /**
     * 最多累计多次进行删除过期节点运行资源信息
     */
    private static final int MAX_CUMULATIVE_NUMBER = 600;
    /**
     * 累积计数器
     */
    private static final AtomicInteger CUMULATIVE = new AtomicInteger();

    private final NodeInfoService nodeInfoService;
    private final NodeUsedInfoService nodeUsedInfoService;
    private final AlarmRuleInfoService alarmRuleInfoService;
    private final AlarmInfoReport alarmInfoReport;
    private final StringRedisTemplate redis;
    private final SysConfig config;

    public NodeUsedPullTask(NodeInfoService nodeInfoService, NodeUsedInfoService nodeUsedInfoService, AlarmRuleInfoService alarmRuleInfoService, AlarmInfoReport alarmInfoReport, StringRedisTemplate redis, SysConfig config) {
        this.nodeInfoService = nodeInfoService;
        this.nodeUsedInfoService = nodeUsedInfoService;
        this.alarmRuleInfoService = alarmRuleInfoService;
        this.alarmInfoReport = alarmInfoReport;
        this.redis = redis;
        this.config = config;
    }

    @Scheduled(cron = "${jobs.map.nodeUsedPull.cron}")
    @SchedulerLock(name = "${jobs.map.nodeUsedPull.name}", lockAtMostFor = "${jobs.map.nodeUsedPull.lockAtMostFor}", lockAtLeastFor = "${jobs.map.nodeUsedPull.lockAtLeastFor}")
    public void execute() throws Exception {
        Date collectTime = new Date();
        Map<Long, NodeInfo> map = nodeInfoService.getCacheNodeInfoAll().stream()
                .filter(o -> Status.NORMAL.code.equals(o.getStatus()) && MONITOR_MODE_PASSIVITY == o.getMonitorMode()).collect(Collectors.toMap(NodeInfo::getId, Function.identity()));
        if (map.isEmpty()) {
            return;
        }
        List<NodeUsedInfo> infoList = map.values().stream()
                .map(o -> getNodeUsedInfo(o, collectTime)).collect(Collectors.toList());
        int size = map.size();
        Map<String, List<Double>> alarmRuleMap = alarmRuleInfoService.getAlarmRuleMap(RULE_TYPE);
        CountDownLatch latch = new CountDownLatch(size);
        for (NodeUsedInfo usedInfo : infoList) {
            NodeInfo info = map.get(usedInfo.getId());
            String url = StringUtil.format("http://{}:{}{}", info.getIp(), info.getPort(), SERVER_INFO_URL);
            OkHttpUtil.get(url, new NodeUsedPullCallback(usedInfo, latch, LOG, alarmInfoReport, alarmRuleMap.get(EVENT_CODE_DISK)));
        }

        boolean isExecuteComplete = latch.await(10L * size, TimeUnit.SECONDS);
        if (!isExecuteComplete) {
            LOG.error("执行节点运行情况收集时间超时");
        }

        List<NodeUsedInfo> infos = infoList.stream().filter(o -> 0 != o.getMemoryTotal()).collect(Collectors.toList());

        for (NodeUsedInfo nodeUsedInfo : infos) {
            nodeUsedInfo.setId(null);
            recordAlarm(nodeUsedInfo, alarmRuleMap);
        }

        nodeUsedInfoService.saveBatch(infos);
        saveNowUsed(infos);
        checkClearUsedInfo(collectTime);
    }

    /**
     * 存入实时的节点使用资源
     *
     * @param infos 信息
     */
    private void saveNowUsed(List<NodeUsedInfo> infos) {
        HashOperations<String, String, String> opsForHash = redis.opsForHash();
        Map<String, String> nowNodeUsedInfoMap = new HashMap<>(infos.size());
        infos.forEach(o -> nowNodeUsedInfoMap.put(String.valueOf(o.getNodeCode()), JsonTool.toJson(o)));
        opsForHash.putAll(REDIS_KEY_NODE_NOW_USED, nowNodeUsedInfoMap);
    }

    /**
     * 检查是否清除节点使用资源信息
     *
     * @param collectTime 收集时间
     */
    private void checkClearUsedInfo(Date collectTime) {
        if (CUMULATIVE.incrementAndGet() > MAX_CUMULATIVE_NUMBER) {
            long millis = config.getNodeUsedRetentionTime().toMillis();
            Calendar calendar = DateUtil.getCalendar(collectTime);
            DateUtil.offsetDate(calendar, DateUtil.MILLISECOND, (int) -millis);
            nodeUsedInfoService.remove(new LambdaQueryWrapper<NodeUsedInfo>()
                    .le(NodeUsedInfo::getDetectionTime, calendar.getTime()));
            CUMULATIVE.set(0);
        }
    }

    /**
     * 获得节点使用资源信息
     *
     * @param info        信息
     * @param collectTime 收集时间
     * @return {@link NodeUsedInfo}
     */
    private NodeUsedInfo getNodeUsedInfo(NodeInfo info, Date collectTime) {
        NodeUsedInfo nodeUsedInfo = new NodeUsedInfo();
        nodeUsedInfo.setId(info.getId());
        nodeUsedInfo.setNodeCode(info.getNodeCode());
        nodeUsedInfo.setHostname(info.getHostname());
        nodeUsedInfo.setIp(info.getIp());
        nodeUsedInfo.setDetectionTime(collectTime);
        nodeUsedInfo.setDiskTotal(0L);
        nodeUsedInfo.setDiskUsed(0L);
        nodeUsedInfo.setDiskUsage(0.0D);
        nodeUsedInfo.setCpuNum(0);
        nodeUsedInfo.setCpuUsed(0.0D);
        nodeUsedInfo.setMemoryTotal(0.0D);
        nodeUsedInfo.setMemoryUsed(0.0D);
        nodeUsedInfo.setNetworkName("");
        nodeUsedInfo.setPacketsReceiveAmount(0L);
        nodeUsedInfo.setPacketsSendAmount(0L);
        nodeUsedInfo.setUploadBps(0L);
        nodeUsedInfo.setDownloadBps(0L);
        nodeUsedInfo.setRegionCode(config.getRegionCode());
        return nodeUsedInfo;
    }

    /**
     * 记录报警
     *
     * @param nodeUsedInfo 节点使用情况信息
     */
    private void recordAlarm(NodeUsedInfo nodeUsedInfo, Map<String, List<Double>> map) {
        Double cpuUsed = nodeUsedInfo.getCpuUsed();
        double memoryUsed = NumberUtil.div(nodeUsedInfo.getMemoryUsed(), nodeUsedInfo.getMemoryTotal(), 8);
        /*double diskUsed = NumberUtil.div(nodeUsedInfo.getDiskUsed().doubleValue(), nodeUsedInfo.getDiskTotal().doubleValue(), 8);*/
        double uploadBps = nodeUsedInfo.getUploadBps().doubleValue();
        double downloadBps = nodeUsedInfo.getDownloadBps().doubleValue();

        recordAlarm(nodeUsedInfo, RULE_TYPE,
                EVENT_CODE_CPU, map.get(EVENT_CODE_CPU), cpuUsed);
        recordAlarm(nodeUsedInfo, RULE_TYPE,
                EVENT_CODE_MEM, map.get(EVENT_CODE_MEM), memoryUsed);
        /*recordAlarm(nodeUsedInfo, RULE_TYPE,
                EVENT_CODE_DISK, map.get(EVENT_CODE_DISK), diskUsed);*/
        recordAlarm(nodeUsedInfo, RULE_TYPE,
                EVENT_CODE_UPLOAD_BPS, map.get(EVENT_CODE_UPLOAD_BPS), uploadBps);
        recordAlarm(nodeUsedInfo, RULE_TYPE,
                EVENT_CODE_DOWNLOAD_BPS, map.get(EVENT_CODE_DOWNLOAD_BPS), downloadBps);
    }

    /**
     * 记录告警
     *
     * @param nodeUsedInfo  节点使用情况信息
     * @param ruleType      规则类型
     * @param eventCode     事件编号
     * @param thresholdList 阈值列表
     * @param used          使用值
     */
    private void recordAlarm(NodeUsedInfo nodeUsedInfo, String ruleType, String eventCode, List<Double> thresholdList, Double used) {
        boolean flag = isExceed(thresholdList, used);
        Integer nodeCode = nodeUsedInfo.getNodeCode();
        if (flag) {
            String alarmContent = StringUtil.format("获取来自IP为 {} 的硬件使用情况，触发了 {} 的告警，当使用率为 {}%", nodeUsedInfo.getIp(), getEventContent(eventCode), NumberUtil.roundStr(used, 4));
            LOG.warn(alarmContent);
            alarmInfoReport.nodeUsedThresholdExceededAlarm(nodeCode, ruleType, eventCode, alarmContent, used);
        } else {
            alarmInfoReport.nodeUsedThresholdExceededClearAlarm(nodeCode, ruleType, eventCode, used);
        }
    }

}

/**
 * 设备节点运行情况收集回调
 *
 * @author SoulBGM
 * @date 2023-01-06
 */
class NodeUsedPullCallback implements Callback {

    private final NodeUsedInfo nodeUsedInfo;
    private final CountDownLatch latch;
    private final LogHelper log;
    private final AlarmInfoReport alarmInfoReport;
    private final List<Double> diskThresholdList;

    public NodeUsedPullCallback(NodeUsedInfo nodeUsedInfo, CountDownLatch latch, LogHelper log, AlarmInfoReport alarmInfoReport, List<Double> diskThresholdList) {
        this.nodeUsedInfo = nodeUsedInfo;
        this.latch = latch;
        this.log = log;
        this.alarmInfoReport = alarmInfoReport;
        this.diskThresholdList = diskThresholdList;
    }

    @Override
    public void onResponse(Call call, Response response) throws IOException {
        ResponseBody body = response.body();
        if (body != null) {
            Server server = JsonTool.parse(body.string(), Server.class);
            if (server == null) {
                return;
            }
            NodeUsedInfo.convert(server, nodeUsedInfo);
            List<SysFile> sysFiles = server.getSysFiles();
            for (SysFile s : sysFiles) {
                diskRecordAlarm(s);
            }
        }
        latch.countDown();
    }

    /**
     * 磁盘超出阈值记录报警
     *
     * @param s 系统文件相关信息
     */
    private void diskRecordAlarm(SysFile s) {
        double used = NumberUtil.div(s.getUsed().doubleValue(), s.getTotal().doubleValue(), 8);

        recordAlarm(nodeUsedInfo,
                diskThresholdList,
                used,
                s.getDirName(),
                alarmInfoReport);
    }


    /**
     * 记录告警
     *
     * @param nodeUsedInfo  节点使用情况信息
     * @param thresholdList 阈值列表
     * @param used          使用值
     * @param dirName       盘符路径
     */
    private void recordAlarm(NodeUsedInfo nodeUsedInfo, List<Double> thresholdList, Double used, String dirName, AlarmInfoReport alarmInfoReport) {
        boolean flag = isExceed(thresholdList, used);
        Integer nodeCode = nodeUsedInfo.getNodeCode();
        if (flag) {
            String alarmContent = StringUtil.format("获取来自IP为 {} 的硬件使用情况，触发了 {} 的告警，当前挂载点 {} 使用率为 {}%", nodeUsedInfo.getIp(), getEventContent(EVENT_CODE_DISK), dirName, NumberUtil.roundStr(used, 4));
            log.warn(alarmContent);
            alarmInfoReport.nodeDiskUsedThresholdExceededAlarm(nodeCode, RULE_TYPE_NODE, EVENT_CODE_DISK, dirName, alarmContent, used);
        } else {
            alarmInfoReport.nodeDiskUsedThresholdExceededClearAlarm(nodeCode, RULE_TYPE_NODE, EVENT_CODE_DISK, dirName, used);
        }
    }

    @Override
    public void onFailure(Call call, IOException e) {
        latch.countDown();
    }

}